package org.apm.data.pusher;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.ClearScrollRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.aggregations.AggregationBuilders;

import com.alibaba.fastjson.JSONObject;

public class UpdateData {

    private static Logger LOG = LogManager.getLogger(UpdateData.class);

    private static final int SCROLL_TIME = 60000;
    private static final int SCROLL_SIZE = 500;

    private static final QueryBuilder pagemonitorQuery = QueryBuilders.boolQuery()
            .mustNot(QueryBuilders.existsQuery("pageUnload"));

    public static void queryUpdate(final Client client, final BulkProcessor processor, final String index,
            QueryBuilder query, Map<String, Object> map) {

        SearchResponse scrollResp = client.prepareSearch(index).setQuery(query).setScroll(new TimeValue(SCROLL_TIME))
                .setSize(SCROLL_SIZE).execute().actionGet();

        ClearScrollRequestBuilder clearRequest = client.prepareClearScroll();
        while (true) {
            // Break condition: No hits are returned
            if (scrollResp.getHits().getHits().length == 0) {
                break;
            } else {
                LOG.info("[Analyzer-After][" + index + "]" + scrollResp.getHits().getHits().length + ".");
            }

            for (SearchHit hit : scrollResp.getHits()) {
                JSONObject json = JSONObject.parseObject(hit.getSourceAsString());
                for (String key : map.keySet()) {
                    json.put(key, map.get(key));
                }

                IndexRequest indexRequest = new IndexRequest(hit.getIndex(), hit.getType(), hit.getId())
                        .source(json.toJSONString());
                processor.add(indexRequest);
            }

            scrollResp = client.prepareSearchScroll(scrollResp.getScrollId()).setScroll(new TimeValue(SCROLL_TIME))
                    .execute().actionGet();
        }

        // 清缓存
        clearRequest.addScrollId(scrollResp.getScrollId()).execute().actionGet().isSucceeded();
        clearRequest.execute().actionGet();

        LOG.info("[Analyzer-After][" + index + "] finished.");
    }

    public static void updatePagemonitor() {
        Map<String, Object> map = new HashMap<>();
        map.put("pageUnload", false);
        queryUpdate(CommonMethod.devClient(), CommonMethod.EsBulkProcessor.DEV.processor(), "apm2.0-*-pagemonitor-1",
                pagemonitorQuery, map);
        queryUpdate(CommonMethod.onlineClient(), CommonMethod.EsBulkProcessor.ONLINE.processor(),
                "apm2.0-*-pagemonitor-1", pagemonitorQuery, map);
    }

    public static void testQuery(final Client client, final String index, QueryBuilder query) {
        System.out.println(client.prepareSearch(index).setQuery(query).execute().actionGet());
    }

    public static void main(String[] args) throws InterruptedException {
        testQuery(CommonMethod.onlineClient(), "apm2.0-*-pagemonitor-1", pagemonitorQuery);
        updatePagemonitor();
        
        Thread.sleep(5000);
    }

    // 关联page,resource,sysmonit
    public static void test(Client client) {
        SearchResponse scrollResp = client.prepareSearch("apm2.0-*-resourcesmonitor-1")
                .setQuery(QueryBuilders.boolQuery().filter(QueryBuilders.rangeQuery("pageResources.duration").to(0)))
                .addStoredField("pageResources.traceId").setScroll(new TimeValue(SCROLL_TIME)).setSize(SCROLL_SIZE).execute()
                .actionGet();

        ClearScrollRequestBuilder clearRequest = client.prepareClearScroll();
        while (true) {
            // Break condition: No hits are returned
            if (scrollResp.getHits().getHits().length == 0) {
                break;
            }

            List<String> traceIds = new ArrayList<>();
            for (SearchHit hit : scrollResp.getHits()) {
                String traceId = hit.getFields().get("pageResources.traceId").getValues().get(0).toString();
                traceIds.add(traceId);
            }

            SearchResponse searchResponse = client.prepareSearch("apm2.0-*-systemmonitor-1")
                    .setQuery(QueryBuilders.boolQuery().filter(QueryBuilders.termsQuery("traceId", traceIds))
                            .filter(QueryBuilders.rangeQuery("timestamp").from("2016-12-13 00:00:00.000")))
                    .setSize(10000).execute().actionGet();

            if (searchResponse.getHits().getTotalHits() > 0) {
                System.out.println(searchResponse);
            }

            scrollResp = client.prepareSearchScroll(scrollResp.getScrollId()).setScroll(new TimeValue(SCROLL_TIME))
                    .execute().actionGet();
        }

        // 清缓存
        clearRequest.addScrollId(scrollResp.getScrollId()).execute().actionGet().isSucceeded();
        clearRequest.execute().actionGet();
    }
}
